package com.tool.o2m;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.google.common.collect.Lists;
import com.tool.o2m.model.Table;
import com.tool.o2m.model.TableColumn;
import com.tool.o2m.model.TableKey;
import com.tool.o2m.model.UnderlineToCamelCaseRowMapper;
import com.tool.util.FileUtils;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * dml
 * oracle to mysql
 */
@Slf4j
public class Oracle2MySQLConverter {

    public void run(String jdbcUrl, String databaseName, String username, String password, Set<String> customSyncTableNames){

        // 使用HikariCP连接池
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl(jdbcUrl);
        config.setUsername(username);
        config.setPassword(password);
        config.addDataSourceProperty("cachePrepStmts", "true");
        config.addDataSourceProperty("prepStmtCacheSize", "125");
        config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
        HikariDataSource ds = new HikariDataSource(config);

        // 使用数据源创建JdbcClient；fluent API的风格的写法，相比JdbcTemplate更简洁
        JdbcClient jdbcClient = JdbcClient.create(ds);

        // 是否全量同步
        boolean isFullSync = CollUtil.isEmpty(customSyncTableNames) ? true : false;
        if (isFullSync) {
            log.info("开始全量同步表结构 ... ！");
        } else {
            log.info("开始自定义同步表结构 ... ！", customSyncTableNames.size());
        }

        // step1 获取每张表的名称和注释
        String getAllTableSql = """
                                select a.owner, a.table_name ,b.comments
                                  from all_tables a
                                  left join all_tab_comments b
                                    on a.owner = b.owner and a.table_name = b.table_name
                                  where a.owner = :databaseName
                                 order by table_name asc
                                """;

        List<Table> allTableList = jdbcClient.sql(getAllTableSql)
                .param(databaseName)
                .query(new UnderlineToCamelCaseRowMapper<>(Table.class)).list();

        if (CollUtil.isEmpty(allTableList)) {
            log.info("没有表需要同步！");
            return;
        }

        List<Table> syncTableList = new ArrayList<>();
        for (Table oracleTable : allTableList) {
            // 判断是全量同步还是自定义同步
            if (isFullSync) {
                syncTableList.add(oracleTable);
            } else {
                if (customSyncTableNames.contains(oracleTable.tableName.toUpperCase())) {
                    syncTableList.add(oracleTable);
                }
            }
        }

        long startTime = System.currentTimeMillis();
        int tableCount = syncTableList.size();
        log.info("开始获取表结构信息 ... ！ 总共{}张表", tableCount);

        //step2 找到每张表的列信息进行DML拼凑和转换
        // 查询表结构
        String tableColumnSql = """
                                    select a.owner
                                         , a.table_name
                                         , a.column_id
                                         , a.column_name
                                         , a.data_type
                                         , case when a.data_type = 'VARCHAR2' or a.data_type = 'NVARCHAR2' then a.char_length else a.data_length end as data_length
                                         , nvl(a.data_precision, 0) as data_precision
                                         , nvl(a.data_scale, 0)     as data_scale
                                         , a.char_length
                                         , a.data_default
                                         , a.nullable
                                         , b.comments
                                         , (sum(case when a.data_type = 'VARCHAR2' or a.data_type = 'NVARCHAR2' then a.char_length else 0 end) over (partition by a.table_name))*4 as varchar_row_size
                                    from all_tab_columns a
                                             inner join all_col_comments b
                                                        on a.owner = b.owner and a.column_name = b.column_name
                                                            and a.table_name = b.table_name
                                    where a.owner = :databaseName and a.table_name in (:tableNames)
                                    order by a.table_name, a.column_id asc                     
                                """;

        // 查询主键和索引；
        String tableKeySql = """
                                with table_index as
                                         (select con.table_name
                                               , con.constraint_name as key_name
                                               , 'PRIMARY'           as key_type
                                               , col.column_name
                                               , null                as column_expression
                                               , nvl(col.position, 1) as column_position
                                          from all_constraints con
                                                   inner join all_cons_columns col
                                                         on con.owner = col.owner and con.table_name = col.table_name 
                                                             and con.constraint_name = col.constraint_name
                                          where con.status = 'ENABLED'
                                            and con.constraint_type = 'P'
                                            and con.owner = :databaseName
                                            and con.table_name in (:tableNames)
                                          union all
                                          select ai.table_name
                                               , ai.index_name as key_name
                                               , ai.uniqueness as key_type
                                               , aic.column_name
                                               , aie.column_expression
                                               , nvl(aic.column_position, 1) as column_position
                                          from all_indexes ai
                                                   inner join all_ind_columns aic 
                                                           on ai.owner = aic.index_owner and ai.table_name = aic.table_name and ai.index_name = aic.index_name
                                                   left join all_ind_expressions aie
                                                          on aic.index_owner = aie.index_owner and aic.table_name = aie.table_name 
                                                         and aic.index_name = aie.index_name and aic.column_position = aie.column_position
                                          where ai.owner = :databaseName
                                            and ai.table_name in (:tableNames)
                                         )
                                select t.*
                                from table_index t
                                order by t.table_name, t.key_type desc, t.key_name, t.column_position
                             """;


        // 按每1000张表进行拆分；因为oracle中in参数最大为1000个。
        // 如果超过1000张表，要分批查询。否则会报错：Caused by: java.sql.SQLSyntaxErrorException: ORA-01795: 列表中的最大表达式数为 1000；
        // 为了性能，按每批300张表进行拆分
        List<List<Table>> tablePartitionList = Lists.partition(syncTableList, 300);

        int x = 0;
        // 定义变量存储所有表的所有字段
        List<TableColumn> columnList = new ArrayList<>();
        List<TableKey> keyList = new ArrayList<>();
        for (List<Table> tables : tablePartitionList) {
            x++;
            log.info("开始查询第{}批表元信息 ---↓↓↓", x);
            System.out.println();

            Set<String> tableNames = tables.stream().map(Table::getTableName).collect(Collectors.toSet());
            Map<String, Object> params = new HashMap<>();
            params.put("databaseName", databaseName);
            params.put("tableNames", tableNames);

            log.info("开始查询列信息");
            List<TableColumn> tableColumns = jdbcClient.sql(tableColumnSql)
                    .paramSource(params)
                    .query(new UnderlineToCamelCaseRowMapper<>(TableColumn.class)).list();

            columnList.addAll(tableColumns);
            log.info("查询列信息结束");
            System.out.println();


            log.info("开始查询键信息");
            List<TableKey> tableKeys = jdbcClient.sql(tableKeySql)
                    .paramSource(params)
                    .query(new UnderlineToCamelCaseRowMapper<>(TableKey.class)).list();

            keyList.addAll(tableKeys);
            log.info("查询键信息结束");
            System.out.println();

            log.info("第{}批表元信息查询结束 ---↑↑↑", x);
            System.out.println();
            System.out.println();
            System.out.println();
        }

        // 按表对列和键进行分组
        Map<String, List<TableColumn>> tableColumnMap = columnList.stream().collect(Collectors.groupingBy(TableColumn::getTableName));
        Map<String, List<TableKey>> tableKeyMap = keyList.stream().collect(Collectors.groupingBy(TableKey::getTableName));

        for(Table table: syncTableList){
            List<TableColumn> tableColumnList = tableColumnMap.get(table.getTableName());
            table.setColumns(tableColumnList);

            List<TableKey> keys = tableKeyMap.get(table.getTableName());
            // 由于column_expression字段是LONG类型，无法在SQL处理数据。所以需要查询出结果后在JAVA中处理
            table.setKeys(keys);

            // 处理 SQL 错误 [1118] [42000]: Row size too large.
            table.varcharRowSize =  tableColumnList.get(0).getVarcharRowSize();
            table.setToTextColumns(tableColumnList);

        }

        //step3 进行转换
        log.info("开始转换表结构 ... ！ 总共{}张表", tableCount);
        StringBuilder sb = new StringBuilder();
        int i = 0;
        for(Table oracleTable: syncTableList){
            i++;
            sb.append(oracleTable.toMySQLScript());
            log.info("第{}张表：{} --> 表结构转换完成 ...... ", i, oracleTable.tableName);
        }

        //step4 save to file
        String scriptComment = "-- " + LocalDateTimeUtil.format(LocalDateTime.now(), DatePattern.NORM_DATETIME_PATTERN)
                + "  Oracle 2 MySQL \n" +
                "-- Database Info: \n" +
                "--     " + jdbcUrl + "\n" +
                "--     " + username + "\n\n\n";
        sb.insert(0, scriptComment);

        log.info("所有表结构转换完成！开始生成脚本文件 ...... ");
        FileUtils.saveSqlScript(sb.toString());
        log.info("DDL脚本保存成功，表结构转换结束！");

        long endTime = System.currentTimeMillis();
        long costTime = endTime - startTime;
        long seconds = costTime / 1000;
        long minutes = seconds / 60;
        long hours = minutes / 60;
        log.info("表结构转换耗时：{}毫秒=={}秒=={}分钟=={}小时", costTime, seconds, minutes, hours);
    }

}
